package com.imooc.bigdata.chapter08_my.business

import com.imooc.bigdata.chapter08_my.`trait`.DataProcess
import com.imooc.bigdata.chapter08_my.utils.KuduUtils
import com.imooc.bigdata.chapter08_my.utils.{DateUtils, SQLUtils, SchemaUtils}
import org.apache.kudu.Schema
import org.apache.spark.sql.{DataFrame, SparkSession}

object AppStatProcess extends DataProcess{
  override def process(spark: SparkSession): Unit = {

    val sourceTableName = DateUtils.getTableName("ods",spark)
    val masterAddress = "spark000"


    val odsDF: DataFrame = spark.read.format("org.apache.kudu.spark.kudu") //
      .option("kudu.table", sourceTableName)
      .option("kudu.master", masterAddress)
      .load()

    odsDF.createOrReplaceTempView("ods")

    val resultTmp: DataFrame = spark.sql(SQLUtils.APP_SQL_STEP1)
    resultTmp.createOrReplaceTempView("app_tmp")

    val resultDF: DataFrame = spark.sql(SQLUtils.APP_SQL_STEP2)
    resultDF.show(false)

    //写回kudu
    val SINK_TABLE_NAME =DateUtils.getTableName("app_stat",spark)
    val partitionId = "appid"
    val schema: Schema = SchemaUtils.APPSchema

    KuduUtils.sink(resultDF,SINK_TABLE_NAME,masterAddress,schema,partitionId)



  }
}
